package com.moter.demo01.rpc;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class SocketFixedWithSelectorIO {
    private ServerSocketChannel server =null;
    private Selector selector1=null;
    private Selector selector2=null;
    private Selector selector3=null;
    private int port = 9090;
    public void initServer()  {
        try {
            server = ServerSocketChannel.open();
            server.bind(new InetSocketAddress(port));
            server.configureBlocking(false);
            selector1 = Selector.open();
            selector2 = Selector.open();
            selector3 = Selector.open();
            server.register(selector1, SelectionKey.OP_ACCEPT);
        }catch (IOException e){
            System.out.println(e.getMessage());
        }
    }
    static class NioThread extends Thread{
        private Selector selector;
        private static LinkedBlockingQueue<SocketChannel> queue[];
        private int id=99;//初始值为master的值
        private boolean isMaster=true;
        private static int selectors = 0;
        private static AtomicInteger idx = new AtomicInteger();
        public NioThread(Selector selector){
            isMaster=false;
            this.selector = selector;
            id = idx.getAndIncrement() % selectors;
            System.out.println("worker "+id+"\t启动");
        }
        public NioThread(Selector selector,int selectors){
            this.selector = selector;
            this.selectors = selectors;
            queue = new LinkedBlockingQueue[this.selectors];
            for (int i = 0; i < this.selectors; i++) {//初始化队列
                queue[i] = new LinkedBlockingQueue<>();
            }
            System.out.println("boss\t 启动");
        }
        @Override
        public void run() {
            try {
                while(true){
                    while (selector.select(10)>0){
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while(iterator.hasNext()){
                            SelectionKey next = iterator.next();
                            iterator.remove();
                            if(next.isAcceptable()){
                                acceptHandler(next);
                            }else if(next.isReadable()){
                                readHandler(next);
                            }
                        }
                    }
                    if(!isMaster&&!queue[id].isEmpty()){ //boss不参与  worker的事情
                        ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
                        SocketChannel client = queue[id].take();
                        client.register(selector,SelectionKey.OP_READ,byteBuffer);
                        System.out.println("===============================");
                        System.out.println("新客户端："+client.socket().getPort()+"\t分配到worker"+id);
                        System.out.println("===============================");
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        private void readHandler(SelectionKey next) {//worker读取信息
            try{
                SocketChannel sc = (SocketChannel) next.channel();
                ByteBuffer byteBuffer = (ByteBuffer) next.attachment();
                byteBuffer.clear();
                int read=0;
                while(true){
                    read = sc.read(byteBuffer);
                    if(read>0){
                        byteBuffer.flip();
                        while(byteBuffer.hasRemaining()){
//                            sc.write(byteBuffer);
                            byte[] data = new byte[byteBuffer.limit()];
                            byteBuffer.get(data);
                            System.out.println("worker"+id+"收到信息：\t"+new String(data));
                            ByteBuffer dataReturn = ByteBuffer.allocate(4096);
                            dataReturn.put(("worker:"+id+"\treturn"+new String(data)).getBytes());
                            dataReturn.flip();
                            sc.write(dataReturn);
                            dataReturn.clear();
                        }
                        byteBuffer.clear();
                    }else if(read==0){
                        break;
                    }else{//read=-1;
                        System.out.println(sc.socket().getPort()+"\t退出连接");
                        sc.close();
                        break;
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void acceptHandler(SelectionKey next) {//boss把获取到的客户端连接给worker做
            try {
                ServerSocketChannel ssc = (ServerSocketChannel)next.channel();
                SocketChannel client = ssc.accept();
                client.configureBlocking(false);
                int num = idx.getAndIncrement()%selectors;
                queue[num].add(client);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SocketFixedWithSelectorIO socketFixedWithSelectorIO = new SocketFixedWithSelectorIO();
        socketFixedWithSelectorIO.initServer();
        NioThread boss = new NioThread(socketFixedWithSelectorIO.selector1, 2);
        NioThread worker1 = new NioThread(socketFixedWithSelectorIO.selector2);
        NioThread worker2 = new NioThread(socketFixedWithSelectorIO.selector3);
        boss.start();
        TimeUnit.SECONDS.sleep(1);
        worker1.start();
        worker2.start();
    }
}
